package com.rocketmq.example;

import java.util.List;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class DefaultPropertyForProducer {

    public static void main(String[] args) throws Exception {
        // producerGroup 生产者所属组，对普通消息意义不大，主要针对事务消息和高可用消息
        // （事务消息中如果某条发送某条消息的producer-A宕机，使得事务消息一直处于PREPARED状态并超时，则broker会回查同一个group的其它producer，确认这条消息应该commit还是rollback）
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(ExampleMQConfig.GROUP.getName());
        // 创建不同的消费者访问不同的MQ集群或服务器时需要使用 （MQ 区分不同生产者的方式是：ip@instanceName@UnitName，前两项都不好修改）
        defaultMQProducer.setUnitName("defaultMQProducerA");
        // 主题在每个 broker 中队列的数量，默认是 4
        defaultMQProducer.setDefaultTopicQueueNums(4);
        // 发送消息的超时时间
        defaultMQProducer.setSendMsgTimeout(3000);
        // 消息体超过该值则进行压缩
        defaultMQProducer.setCompressMsgBodyOverHowmuch(1024 * 4);
        // 同步发送消息重试次数，默认两次，一共发送三次
        defaultMQProducer.setRetryTimesWhenSendFailed(2);
        // 异步发送消息重试次数，默认两次，一共发送三次
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(2);
        // 消息没有发送成功是否发送到另外一个 broker
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(false);
        // 允许发送消息的最大长度，默认为 4M
        defaultMQProducer.setMaxMessageSize(1024 * 1024 * 4);
        // 设置 NameServer 地址
        defaultMQProducer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());
        // 采用延迟规避策略，适合网络不好的情况，避免直接规避由于一瞬间网络异常导致程序异常的部分服务器
        defaultMQProducer.setSendLatencyFaultEnable(true);
        // 启动生产者
        defaultMQProducer.start();
        // 查询主题下的所有队列
        List<MessageQueue> fetchPublishMessageQueueList = defaultMQProducer
                .fetchPublishMessageQueues(ExampleMQConfig.TOPIC.getName());
        fetchPublishMessageQueueList.forEach(fetchPublishMessageQueue -> {
            System.err.println(fetchPublishMessageQueue.getQueueId());
        });

        Message message = new Message(ExampleMQConfig.TOPIC.getName(), ExampleMQConfig.TAG.getName(), "key1",
                "message".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 单向发送
        // 单向发送消息，不关心返回，这种方式发送消息最快，但是不可信，适合不要求准确的场景，比如日志记录，丢失几条日志是可以接受的
        defaultMQProducer.sendOneway(message);
        // 指定队列单向发送消息
        defaultMQProducer.sendOneway(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return mqs.get(0);
            }
        }, null);
        // 指定队列单向发送消息（根据之前查出来的主题）
        defaultMQProducer.sendOneway(message, fetchPublishMessageQueueList.get(0));

        // 同步发送
        // 同步发送消息
        SendResult sendResult = defaultMQProducer.send(message);
        SendStatus sendStatus = sendResult.getSendStatus();
        System.err.println(sendStatus.toString());
        // 发送超时时间设置
        defaultMQProducer.send(message, 1000 * 3);
        // 指定队列发送同步消息
        defaultMQProducer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return mqs.get(0);
            }
        }, null);

        // 指定队列发送同步消息（根据之前查出来的主题）
        defaultMQProducer.send(message, fetchPublishMessageQueueList.get(0));

        // 异步发送
        //  异步发送消息
        defaultMQProducer.send(message,new SendCallback() {

            // 发送成功
            @Override
            public void onSuccess(SendResult sendResult) {
                // TODO Auto-generated method stub
                throw new UnsupportedOperationException("Unimplemented method 'onSuccess'");
            }

            // 发送失败
            @Override
            public void onException(Throwable e) {
                // TODO Auto-generated method stub
                throw new UnsupportedOperationException("Unimplemented method 'onException'");
            }
            // 指定超时时间
        },1000 * 3);


        Thread.sleep(10 * 1000);

        defaultMQProducer.shutdown();


    }

}
